package io.plus.interactive.task;

import drds.common.Clock;
import io.plus.interactive.AsynchronousSocketChannelContext;
import io.plus.interactive.CloseCode;
import io.plus.interactive.Configuration;
import io.plus.interactive.Io;
import io.plus.interactive.exception.AioDecodeException;
import io.plus.interactive.state.ChannelState;
import io.plus.interactive.user_interface.Packet;
import io.plus.interactive.utils.ByteBufferUtils;
import io.plus.utils.queue.Queue;
import io.plus.utils.queue.QueueImpl;
import io.plus.utils.thread.lock_capable_task.AbstractQueueTask;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;

/**
 * 解码任务对象，一个连接对应一个本对象
 */
@Data
@Slf4j
public class DecodeTask extends AbstractQueueTask<ByteBuffer> {
    private AsynchronousSocketChannelContext asynchronousSocketChannelContext = null;
    private Configuration configuration = null;
    private Queue<ByteBuffer> queue = null;
    /**
     * 上一次解码剩下的数据
     */
    private ByteBuffer lastDecodeLeftByteBuffer = null;
    /**
     * 新收到的数据
     */
    private ByteBuffer newReceivedByteBuffer = null;


    public DecodeTask(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Executor executor) {
        super(executor);
        this.asynchronousSocketChannelContext = asynchronousSocketChannelContext;
        this.configuration = asynchronousSocketChannelContext.clientConfiguration;
        getQueue();
    }

    @Override
    public Queue<ByteBuffer> getQueue() {
        if (configuration.useQueueDecode) {
            if (queue == null) {
                synchronized (this) {
                    if (queue == null) {
                        queue = new QueueImpl<ByteBuffer>();
                    }
                }
            }
            return queue;
        }
        return null;
    }

    /**
     * 清空处理的队列消息
     */
    public void clearQueue() {
        super.clearQueue();
        lastDecodeLeftByteBuffer = null;
        newReceivedByteBuffer = null;
    }

    @Override
    public void runTask() {
        while ((newReceivedByteBuffer = queue.poll()) != null) {
            decode();
        }
    }

    public void decode() {
        ByteBuffer byteBuffer = newReceivedByteBuffer;
        if (lastDecodeLeftByteBuffer != null) {
            byteBuffer = ByteBufferUtils.composite(lastDecodeLeftByteBuffer, byteBuffer);//组合为一个数据包
            lastDecodeLeftByteBuffer = null;
        }
        goOnDecode:
        while (true) {
            try {
                int position = byteBuffer.position();
                int limit = byteBuffer.limit();
                int readableLength = limit - position;
                //
                Packet packet = null;
                if (asynchronousSocketChannelContext.packetNeededLength != null) {
                    if (log.isInfoEnabled()) {
                        log.info("{}, 解码所需长度:{}", asynchronousSocketChannelContext, asynchronousSocketChannelContext.packetNeededLength);
                    }
                    if (readableLength >= asynchronousSocketChannelContext.packetNeededLength) {
                        packet = configuration.getAioHandler().decode(asynchronousSocketChannelContext, byteBuffer, limit, position, readableLength);
                    } else {
                        //需要继续等待
                    }
                } else {
                    try {
                        packet = configuration.getAioHandler().decode(asynchronousSocketChannelContext, byteBuffer, limit, position, readableLength);
                    } catch (BufferUnderflowException e) {
                        //数据不够读
                    }
                }

                if (packet == null) {
                    // 数据不够，解不了码
                    if (configuration.useQueueDecode || (byteBuffer != newReceivedByteBuffer)) {
                        byteBuffer.position(position);
                        byteBuffer.limit(limit);
                        lastDecodeLeftByteBuffer = byteBuffer;
                    } else {
                        lastDecodeLeftByteBuffer = ByteBufferUtils.copy(byteBuffer, position, limit);
                    }
                    ChannelState channelState = asynchronousSocketChannelContext.channelState;
                    channelState.decodeFailCount++;
                    if (log.isInfoEnabled()) {
                        log.info("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", asynchronousSocketChannelContext, channelState.decodeFailCount, readableLength);
                    }
                    if (channelState.decodeFailCount > 5) {
                        if (asynchronousSocketChannelContext.packetNeededLength == null) {
                            if (log.isInfoEnabled()) {
                                log.info("{} 本次解码失败, 已经连续{}次解码失败，参与解码的数据长度共{}字节", asynchronousSocketChannelContext, channelState.decodeFailCount, readableLength);
                            }
                        }

                        //检查慢包攻击
                        if (channelState.decodeFailCount > 10) {
                            int per = readableLength / channelState.decodeFailCount;
                            if (per < Math.min(asynchronousSocketChannelContext.getReadBufferSize() / 2, 256)) {
                                String str = "连续解码" + channelState.decodeFailCount + "次都不成功，并且平均每次接收到的数据为" + per + "字节，有慢攻击的嫌疑";
                                throw new AioDecodeException(str);
                            }
                        }
                    }
                    return;
                } else {//解码成功
                    asynchronousSocketChannelContext.setPacketNeededLength(null);
                    asynchronousSocketChannelContext.channelState.latestTimeOfReceivedPacket = Clock.currentTimeMillis;
                    asynchronousSocketChannelContext.channelState.decodeFailCount = 0;

                    int packetSize = byteBuffer.position() - position;
                    packet.setByteCount(packetSize);


                    if (configuration.getAioListener() != null) {
                        try {
                            configuration.getAioListener().onAfterDecoded(asynchronousSocketChannelContext, packet, packetSize);
                        } catch (Throwable e) {
                            log.error(e.toString(), e);
                        }
                    }

                    if (log.isDebugEnabled()) {
                        log.debug("{}, 解包获得一个packet:{}", asynchronousSocketChannelContext, packet.toString());
                    }

                    handler(packet, packetSize);

                    if (byteBuffer.hasRemaining())//组包后，还剩有数据
                    {
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，还剩有数据:{}", asynchronousSocketChannelContext, byteBuffer.remaining());
                        }
                        continue goOnDecode;
                    } else//组包后，数据刚好用完
                    {
                        lastDecodeLeftByteBuffer = null;
                        if (log.isDebugEnabled()) {
                            log.debug("{},组包后，数据刚好用完", asynchronousSocketChannelContext);
                        }
                        return;
                    }
                }
            } catch (Throwable e) {
                log.error("解码异常", e);
                asynchronousSocketChannelContext.setPacketNeededLength(null);
                Io.close(asynchronousSocketChannelContext, e, "解码异常:" + e.getMessage(), CloseCode.DECODE_ERROR);
                return;
            }
        }
    }

    //
    public void handler(Packet packet, int byteCount) {
        switch (configuration.packetHandlerMode) {
            case SINGLE_THREAD:
                asynchronousSocketChannelContext.handlerRunnable.handler(packet);
                break;
            case QUEUE:
                asynchronousSocketChannelContext.handlerRunnable.add(packet);
                asynchronousSocketChannelContext.handlerRunnable.execute();
                break;
            default:
                asynchronousSocketChannelContext.handlerRunnable.handler(packet);
                break;
        }
    }
}
